These changes are necessary to provide enough resources for UCX. For reference, we already do the same for UCXX and RapidsMPF, both need it for the same reason:

• https://github.com/rapidsai/rapidsmpf/blob/14569731349e832dfc12e2105bf5cfe7515ca459/.github/workflows/pr.yaml#L219
• https://github.com/rapidsai/ucxx/blob/e87e2c52d634562c8f2dcffb6488ea5c9afb624a/.github/workflows/pr.yaml#L284

Fixed a bug in the conditional-join fallback logic. We need to repartition the smaller table to a single broadcastable partition whenever either side has more than one partition.

Previously the fallback only triggered when the smaller side itself had multiple partitions, which breaks asymmetric cases like (left=2, right=1). In that situation the single right partition only exists on one rank, so peer ranks execute the conditional join against an empty right side and silently drop rows.

@TomAugspurger and @rjzamora, does that sound correct?

I'll defer to Rick on this...

Though if the new way is correct, then maybe the logic would be a bit more obvious if we combine the first two if conditions?

if (left_count < right_count) and (output_count > 1 or dynamic_planning): 
    left = Repartition(left.schema, left)
    pi_left[left] = PartitionInfo(count=1)
    _fallback_inform(fallback_msg, config_options)
elif output_count > 1 or dynamic_planning:
    right = Repartition(right.schema, right)
    pi_right[right] = PartitionInfo(count=1)
    _fallback_inform(fallback_msg, config_options)

Then the repeated condition (left_count < right_count) will maybe a bit a bit clear (that's true, but the second condition output_count > 1 or dynamic_planning isn't.

At least I hope those are logically equivalent.

I guess we are maybe disabling dynamic planning for some join tests? By default, dynamic-planning is always on, so we always repartition the smaller side.

I guess we are maybe disabling dynamic planning for some join tests? By default, dynamic-planning is always on, so we always repartition the smaller side.

Maybe, it only happens when running with multiple GPUs

Though if the new way is correct, then maybe the logic would be a bit more obvious if we combine the first two if conditions?

Yes, and we can even simplify a bit more. Done

In conda/recipes/cudf-polars/recipe.yaml, we'll probably want to add a

  run_constraints:
    - ray >=2.55.1

to mirror this constraint

Ray is still optional, this is just for the CI testing

Right, IIRC run-constraints is like specifying a version constraint for optional runtime dependencies (like what got added to the pyproject.toml)

https://rattler-build.prefix.dev/latest/reference/recipe_file/#run-constraints

Packages that are optional at runtime but must obey the supplied additional constraint if they are installed.

But can be done in a follow up since the CI is currently green

Ah, I didn't know that, thanks.
But yes, let's do it in a follow-up PR.

Sure thing, opened #22414